package com.chunfen.tomcat.nio.reactor;

import com.fasterxml.jackson.annotation.JsonInclude;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class SubReactor extends Thread{

    private ObjectMapper objectMapper = new ObjectMapper().setSerializationInclusion(JsonInclude.Include.NON_NULL);

    private Selector selector;

    ThreadPoolExecutor pool = new ThreadPoolExecutor(3, 5, 1000,TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(100), new ThreadPoolExecutor.AbortPolicy());

    public SubReactor(){
        try {
            selector = Selector.open();
            System.out.println(Thread.currentThread().getName() + " SubReactor selector open");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void register(SocketChannel socketChannel){
        try {
            System.out.println(Thread.currentThread().getName() + " SubReactor register socketChannel");
            // 一个坑 如果 selector 优先 被select() 方法阻塞， 要优先调用 wakeup()方法，再进行 register 的动作
//            selector.wakeup();
//            new ReadHandler(selector, socketChannel);
            pool.execute(new ReadHandler(selector, socketChannel));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void run() {
        try {
            System.out.println(Thread.currentThread().getName() + " SubReactor started");
            while(!Thread.interrupted()){
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                //Selector如果发现channel有OP_ACCEPT或READ事件发生，下列遍历就会进行。
                while(iterator.hasNext()){

                    //来一个事件 第一次触发一个accepter线程
                    //以后触发SocketReadHandler
                    SelectionKey selectionKey = iterator.next();
                    System.out.println(Thread.currentThread().getName() + " SubReactor from client select" + objectMapper.writeValueAsString(selectionKey));

                    //所有的关心的 io 事件 都 通过 dispatch 进行分发
                    dispatch(selectionKey);
                    //                selectionKeys.clear();
                    iterator.remove();
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println(Thread.currentThread().getName() + " SubReactor run end");
    }

    /**
     * 运行Acceptor或SocketReadHandler
     * @param key
     */
    void dispatch(SelectionKey key) {
        // 不同的 key 触发不同的 事件
        // 注意 此处 attachment 不一定 必须是 Runnable 接口的实现，也可以是公共的接口
        Runnable r = (Runnable)(key.attachment());
        if (r != null){
            r.run();
        }
    }
}
